package mr

import (
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

// 定义为全局，worker之间访问coordinator时加锁
var (
	mu sync.Mutex
)

type Coordinator struct {
	// Your definitions here.
	ReducerNum        int      //Reducer数量
	TaskPhase         Phase    //任务整体状态
	Files             []string //传入文件
	TaskChannelMap    chan *Task
	TaskChannelReduce chan *Task
	TaskId            int            //用于通过自增生成任务id
	TaskMetaHolderMap TaskMetaHolder //保存所有任务
}
type TaskInfo struct {
	state     State     //任务状态
	startTime time.Time //开始时间
	task      *Task     //任务
}

type TaskMetaHolder struct {
	TaskMap map[int]*TaskInfo //保存所有任务
}

// Your code here -- RPC handlers for the worker to call.

//
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
func (c *Coordinator) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}

//
// start a thread that listens for RPCs from worker.go
//
func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := coordinatorSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
//
func (c *Coordinator) Done() bool {
	ret := false

	// Your code here.
	if c.TaskPhase == AllDone {
		ret = true
	}
	return ret
}

//
// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{
		ReducerNum:        nReduce,
		TaskPhase:         MapPhase,
		Files:             files,
		TaskChannelMap:    make(chan *Task, len(files)),
		TaskChannelReduce: make(chan *Task, nReduce),
		TaskMetaHolderMap: TaskMetaHolder{
			TaskMap: make(map[int]*TaskInfo, len(files)+nReduce),
		},
	}

	// Your code here.
	c.makeMapTasks(files)
	c.server()
	return &c
}

//创建map任务
func (c *Coordinator) makeMapTasks(files []string) {
	for _, file := range files {
		fmt.Println(file)
		mtask := Task{
			TaskType:   MapTask,
			FileName:   file,
			ReducerNum: c.ReducerNum,
			TaskId:     c.getTaskId(),
		}

		log.Println("make new map task" + strconv.Itoa(mtask.TaskId))
		c.TaskMetaHolderMap.acceptTask(&mtask)
		c.TaskChannelMap <- &mtask
	}
}

//创建reduce任务
func (c *Coordinator) makeReduceTasks() {
	for i := 0; i < c.ReducerNum; i++ {
		files := getReduceFiles(i)
		rTask := Task{
			TaskType:   ReduceTask,
			TaskId:     c.getTaskId(),
			FileSlice:  files,
			ReducerNum: c.ReducerNum,
		}
		log.Println("make new reduce task" + strconv.Itoa(rTask.TaskId))
		c.TaskMetaHolderMap.acceptTask(&rTask)
		c.TaskChannelReduce <- &rTask
	}
}

//保存新任务
func (t *TaskMetaHolder) acceptTask(task *Task) {
	t.TaskMap[task.TaskId] = &TaskInfo{
		task:  task,
		state: Waiting,
	}
}
func (c *Coordinator) getTaskId() int {
	id := c.TaskId
	c.TaskId++
	return id
}

//获取reduce任务的输入文件
func getReduceFiles(i int) []string {
	files, err := ioutil.ReadDir(".")
	filesName := make([]string, 0)
	if err != nil {
		fmt.Println(err)
		return nil
	}
	for _, file := range files {
		if !file.IsDir() && strings.HasPrefix(file.Name(), "mr-tmp") && strings.HasSuffix(file.Name(), strconv.Itoa(i)) {
			filesName = append(filesName, file.Name())
		}
	}
	return filesName
}

//分配任务
func (c *Coordinator) PollTask(args *TaskArgs, reply *Task) error {
	//上锁防止竞争
	mu.Lock()
	defer mu.Unlock()
	switch c.TaskPhase {
	case MapPhase:
		{
			if len(c.TaskChannelMap) > 0 {
				//分配任务
				*reply = *<-c.TaskChannelMap
				//修改任务状态 记录任务开始时间
				c.TaskMetaHolderMap.TaskMap[reply.TaskId].state = Working
				c.TaskMetaHolderMap.TaskMap[reply.TaskId].startTime = time.Now()
			} else {
				//没有任务 说明所有map任务还在执行中
				reply.TaskType = WaitingTask
				//检查所有任务状态 如果完成则更新进度
				c.checkTimeOver()
				if c.TaskMetaHolderMap.checkTaskDone() {
					c.updateTaskPhase()
				}
			}
		}
	case ReducePhase:
		{
			if len(c.TaskChannelReduce) > 0 {
				//分配reduce任务
				*reply = *<-c.TaskChannelReduce
				//修改任务状态 记录任务开始时间
				c.TaskMetaHolderMap.TaskMap[reply.TaskId].state = Working
				c.TaskMetaHolderMap.TaskMap[reply.TaskId].startTime = time.Now()
			} else {
				reply.TaskType = WaitingTask
				c.checkTimeOver()
				if c.TaskMetaHolderMap.checkTaskDone() {
					c.updateTaskPhase()
				}
			}
		}
	default:
		{
			reply.TaskType = ExitTask
		}
	}
	return nil
}

//检查是否有超时任务
func (c *Coordinator) checkTimeOver() {
	for _, taskInfo := range c.TaskMetaHolderMap.TaskMap {
		if taskInfo.state == Working && time.Since(taskInfo.startTime).Seconds() > 10 {
			taskInfo.state = Waiting
			if taskInfo.task.TaskType == MapTask {
				c.TaskChannelMap <- taskInfo.task
			} else if taskInfo.task.TaskType == ReduceTask {
				c.TaskChannelReduce <- taskInfo.task
			}
		}
	}
}

//检查所有任务完成状态
func (t *TaskMetaHolder) checkTaskDone() bool {
	var (
		mapDoneNum      = 0
		mapUnDoneNum    = 0
		reduceDoneNum   = 0
		reduceUnDoneNum = 0
	)
	for _, taskInfo := range t.TaskMap {
		if taskInfo.task.TaskType == MapTask {
			if taskInfo.state == Done {
				mapDoneNum++
			} else {
				mapUnDoneNum++
			}
		} else if taskInfo.task.TaskType == ReduceTask {
			if taskInfo.state == Done {
				reduceDoneNum++
			} else {
				reduceUnDoneNum++
			}
		}
	}
	if mapUnDoneNum == 0 && reduceUnDoneNum == 0 {
		return true
	}
	return false
}

//更新总任务状态到下一步
func (c *Coordinator) updateTaskPhase() {
	if c.TaskPhase == MapPhase {
		c.TaskPhase = ReducePhase
		//创建reduce任务
		c.makeReduceTasks()
	} else if c.TaskPhase == ReducePhase {
		c.TaskPhase = AllDone
	}
}

//Rpc方法更新任务状态
func (c *Coordinator) TaskDone(args *Task, reply *Task) error {
	// switch args.TaskType {
	// case MapTask:
	// 	{
	meta, ok := c.TaskMetaHolderMap.TaskMap[args.TaskId]
	if ok && meta.state == Working {
		meta.state = Done
		fmt.Printf("task id Id[%d] is finished!\n", args.TaskId)
	} else {
		fmt.Printf("task id Id[%d] is already finished!\n", args.TaskId)
	}
	// 	}
	// }
	return nil
}
